package com.zshield.proc;

import com.google.gson.Gson;
import com.google.gson.JsonObject;
import com.google.gson.JsonParser;
import com.zshield.config.KafkaConfig;
import com.zshield.entry.*;
import com.zshield.run.KafkaPreCompute;
import com.zshield.util.HealthMonitor;
import org.apache.kafka.streams.processor.AbstractProcessor;
import org.apache.kafka.streams.processor.ProcessorContext;
import org.apache.kafka.streams.processor.PunctuationType;
import org.apache.log4j.Logger;

import java.util.HashSet;
import java.util.Set;

public class PreProc extends AbstractProcessor<String, String> {
    private ProcessorContext context;
    private JsonParser jp;
    private Gson gson;
    private int count;
    private JsonObject obj;
    private Set<String> actSet;
    private Set<String> accessSet;
    private final Logger logger = Logger.getLogger(PreProc.class);

    @Override
    public void init(ProcessorContext context) {
        this.context = context;
        jp = new JsonParser();
        gson = new Gson();

        actSet = new HashSet<>(7);
        actSet.add("Read");
        actSet.add("Write");
        actSet.add("Move");
        actSet.add("Create");
        actSet.add("Copy");
        actSet.add("Delete");
        actSet.add("Rename");

        accessSet = new HashSet<>(6);
        accessSet.add("Read");
        accessSet.add("Write");
        accessSet.add("Move");
        accessSet.add("Create");
        accessSet.add("Copy");
        accessSet.add("Rename");
        this.context.schedule(60 * 1000, PunctuationType.WALL_CLOCK_TIME, timestamp -> {
            if (count > 0) {
                logger.info("[stream calculate speed] [Count:" + count + "|Time:60s|Speed:" + String.format("%d", count / 60) + "]");
                count = 0;
                this.context.commit();
            }
        });

        this.context.schedule(3 * 1000, PunctuationType.WALL_CLOCK_TIME, timestamp -> {
            HealthMonitor.ChangingTimeMillis = System.currentTimeMillis();
        });
    }

    @Override
    public void process(String key, String value) {
        try {
            obj = jp.parse(value).getAsJsonObject();
            String format = obj.get("FORMAT").getAsString();
            count++;
            switch (format) {
                case "SENSOR_NETIO":
                    if (obj.get("SENSOR_ID").getAsString().startsWith("LNX")) {
                        if (!obj.get("FILE_HASH").getAsString().equals("")) {
                            LinuxNetioEntry linuxNetIo = new LinuxNetioEntry(obj);
                            context.forward(linuxNetIo.getDocId(), gson.toJson(linuxNetIo), "linux_netio_query_proc");
                        }
                    }
                    NetIoEntry netIo = new NetIoEntry(obj);
                    context.forward(netIo.getDocId(), gson.toJson(netIo), "netio_query_proc");
                    break;
                case "SENSOR_PRINT":
                    FileSystemInOutEntry printInOut = new FileSystemInOutEntry(obj);
                    context.forward(printInOut.getDocId(), gson.toJson(printInOut), "filesystem_inout_query_proc");
                    break;
                case "SENSOR_NET_ACCESS_DENY":
                case "SENSOR_NETWORK_ABNORMAL":
                    NetDenyEntry netDeny = new NetDenyEntry(obj);
                    context.forward(netDeny.getDocId(), gson.toJson(netDeny), "net_deny_query_proc");
                    break;
                case "SENSOR_FILESYSTEM":
                    String fsAction = obj.get("FS_ACTION").getAsString();
                    String accessFormat = judgeAccessFormat(fsAction);
                    String fg = obj.get("FILE_GUID").getAsString();
                    if (actSet.contains(fsAction)) {
                        FileSystemEntry fileSystem = new FileSystemEntry(obj, fsAction, accessSet,accessFormat);
                        context.forward(fileSystem.getDocId(), gson.toJson(fileSystem), "filesystem_query_proc");
                    }
                    if (!"".equals(fg)) {
                        FileSystemTreeEntry fileTree = new FileSystemTreeEntry(obj, fsAction, accessFormat, fg);
                        context.forward(fileTree.getDocId(), gson.toJson(fileTree), "filetree_query_proc");
                    }
                    if (KafkaPreCompute.is_upgrade && !(fsAction.equals("Delete") || fsAction.equals("Execute") || fsAction.equals("Exit"))) {
                        FileSystemInOutEntry fileSystemInOut = new FileSystemInOutEntry(obj, fsAction,accessFormat, true);
                        context.forward(fileSystemInOut.getDocId(), gson.toJson(fileSystemInOut), "filesystem_inout_query_proc");
                    } else {
                        if (!"".equals(accessFormat)) {
                            FileSystemInOutEntry fileSystemInOut = new FileSystemInOutEntry(obj, accessFormat);
                            context.forward(fileSystemInOut.getDocId(), gson.toJson(fileSystemInOut), "filesystem_inout_query_proc");
                        }
                    }
                    break;
                case "SENSOR_FILEPARSE_LOG":
                    ContentAnalysisEntry contentanalysis = new ContentAnalysisEntry(obj);
                    context.forward(contentanalysis.getDocId(), gson.toJson(contentanalysis), "content_analysis_query_proc");
                    break;
            }
        } catch (Throwable e) {
            logger.error("[PreProc process exception] [The reason for error {" + e + "}]");
            KafkaConfig.printErrorLog(logger,e);
        }
    }

    public String judgeAccessFormat(String fsAction) {
        String accessFormat = "";
        if (KafkaPreCompute.is_upgrade) {
            String deviceType = obj.get("DEVICE_TYPE").getAsString();
            String deviceType2 = obj.get("DEVICE_TYPE2").getAsString();
            String userName = obj.get("USERNAME").getAsString();
            String exeName = obj.get("EXE_NAME").getAsString();
            if (deviceType.equals("Fixed") && deviceType2.equals("") && fsAction.equals("Read")) {
                accessFormat = "LOCAL";
            } else if (deviceType.equals("Fixed") && deviceType2.equals("") && fsAction.equals("Write")) {
                accessFormat = "LOCAL";
            } else if (deviceType.equals("Fixed") && deviceType2.equals("Fixed")) {
                accessFormat = "LOCAL";
            } else if (deviceType.equals("Fixed") && deviceType2.equals("Remote")) {
                accessFormat = "SHARE_OUT";
            } else if (deviceType.equals("Fixed") && deviceType2.equals("Removable") ) {
                accessFormat = "USB_OUT";
            } else if (deviceType.equals("Fixed") && deviceType2.equals("CdDvd")) {
                accessFormat = "CD_OUT";
            } else if (deviceType.equals("Removable") && deviceType2.equals("") && fsAction.equals("Read")) {
                accessFormat = "USB_IN";
            } else if (deviceType.equals("Removable") && deviceType2.equals("") && fsAction.equals("Write")) {
                accessFormat = "USB_OUT";
            } else if (deviceType.equals("Removable") && deviceType2.equals("Fixed")) {
                accessFormat = "USB_IN";
            } else if (deviceType.equals("Removable") && deviceType2.equals("Remote")) {
                accessFormat = "USB_IN";
            } else if (deviceType.equals("Removable") && deviceType2.equals("Removable")) {
                accessFormat = "EXTERNAL";
            } else if (deviceType.equals("Removable") && deviceType2.equals("CdDvd")) {
                accessFormat = "EXTERNAL";
            } else if (deviceType.equals("Remote") && deviceType2.equals("") && fsAction.equals("Read")) {
                accessFormat = "SHARE_IN";
            } else if (deviceType.equals("Remote") && deviceType2.equals("") && fsAction.equals("Write")) {
                accessFormat = "SHARE_OUT";
            } else if (deviceType.equals("Remote") && deviceType2.equals("Fixed")) {
                accessFormat = "SHARE_IN";
            } else if (deviceType.equals("Remote") && deviceType2.equals("Removable")) {
                accessFormat = "USB_OUT";
            } else if (deviceType.equals("Remote") && deviceType2.equals("Remote")) {
                accessFormat = "EXTERNAL";
            } else if (deviceType.equals("Remote") && deviceType2.equals("CdDvd")) {
                accessFormat = "CD_OUT";
            } else if (deviceType.equals("CdDvd") && deviceType2.equals("") && fsAction.equals("Read")) {
                accessFormat = "CD_IN";
            } else if (deviceType.equals("CdDvd") && deviceType2.equals("") && fsAction.equals("Write")) {
                accessFormat = "CD_OUT";
            } else if (deviceType.equals("CdDvd") && deviceType2.equals("Fixed")) {
                accessFormat = "CD_IN";
            } else if (deviceType.equals("CdDvd") && deviceType2.equals("Removable")) {
                accessFormat = "EXTERNAL";
            } else if (deviceType.equals("CdDvd") && deviceType2.equals("Remote")) {
                accessFormat = "CD_IN";
            } else if (deviceType.equals("CdDvd") && deviceType2.equals("CdDvd")) {
                accessFormat = "EXTERNAL";
            } else if (exeName.equals("system") && !userName.equals("system")) {
                accessFormat = "SHARE_OUT";
            }
        } else {
            accessFormat = obj.get("ACCESS_FORMAT").getAsString();
        }
        return accessFormat;
    }
}
